package com.sxu.douban.mapreduce;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class Step2 {
    public static class mymapper extends Mapper<Object, Text, Text, Text> {
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString().split("\\s+")[1];
            String line1 =line.substring(0, line.length()-1);
            String[]  movie_list=line1.split(",");
            int len=movie_list.length;
            for(int i=0;i<len;i++){
                for(int j=i+1;j<len;j++){
                    String A=movie_list[i];
                    String B=movie_list[j];
                    String a=A.split(":")[0].split("@")[0];
                    String b=A.split(":")[1];
                    String c=B.split(":")[0].split("@")[0];
                    String d=B.split(":")[1];
                    if(Integer.parseInt(a)>Integer.parseInt(c)){
                        context.write(new Text(B.split(":")[0]+","+A.split(":")[0]), new Text(d+","+b));
                    }
                    else{
                        context.write(new Text(A.split(":")[0]+","+B.split(":")[0]), new Text(b+","+d));
                    }
                }
            }
        }
    }

    public static class myreducer extends Reducer<Text, Text, Text, Text> {
        public void reduce(Text key, Iterable<Text> values, Context context)
                throws InterruptedException, IOException {
            String temp="";
            for(Text value:values){
                temp+=value.toString()+";";
            }
            context.write(key, new Text(temp));
        }
    }

    public static void run() throws IOException, ClassNotFoundException, InterruptedException {
        Configuration configuration = new Configuration();
        configuration.set("fs.defaultFS","hdfs://192.168.22.131:9000");
        FileSystem fileSystem = FileSystem.get(configuration);
        Job job = Job.getInstance(configuration);
        job.setJarByClass(Step2.class);
        job.setMapperClass(Step2.mymapper.class);
        job.setReducerClass(Step2.myreducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, new Path("/Movie_Recommend/output1"));
        Path outPath = new Path("/Movie_Recommend/output2");
        if (fileSystem.exists(outPath)){
            fileSystem.delete(outPath,true);
        }
        FileOutputFormat.setOutputPath(job, outPath);
        job.waitForCompletion(true);
    }

}

